package websocket

import (
	"context"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/gcron"
	"github.com/gogf/gf/v2/os/gtime"
	"runtime/debug"
	"sync"
	"time"
)

type Hub struct {
	clients     map[int64][]*Client //全部的连接，map.key 为用户ID，重复登录后覆盖，保证用户唯一
	lock        sync.Mutex
	register    chan *Client     //客户端连接处理
	unregister  chan *Client     //客户端断开连接处理
	broadcast   chan *WsResponse //广播消息：发送给全员用户
	closeSignal chan struct{}    //websocket服务关闭信号
}

func NewHub() *Hub {
	return &Hub{
		clients:     make(map[int64][]*Client),
		register:    make(chan *Client, 1000),
		unregister:  make(chan *Client, 1000),
		broadcast:   make(chan *WsResponse, 1000),
		closeSignal: make(chan struct{}, 1),
	}
}

// handleRegister 客户端连接注册
func (h *Hub) handleRegister(c *Client) {
	h.lock.Lock()
	defer func() {
		h.lock.Unlock()
	}()

	h.clients[c.user.Id] = append(h.clients[c.user.Id], c)
	Logger().Infof(ctx, "user connect success---clientID=[%v]userINFO[%d,%s]", c.id, c.user.Id, c.user.Username)
}

// 客户端退出连接
func (h *Hub) handleUnregister(c *Client) {
	h.lock.Lock()
	defer func() {
		h.lock.Unlock()
	}()

	if _, ok := h.clients[c.user.Id]; ok {
		delete(h.clients, c.user.Id)
		c.close() //客户端断开连接
	}
}

func (h *Hub) start() {
	defer func() {
		if r := recover(); r != nil {
			Logger().Errorf(ctx, "websocket start error：%+v", string(debug.Stack()))
			return
		}
	}()
	for {
		select {
		case client := <-h.register:
			// 处理用户注册事件
			h.handleRegister(client)
		case client := <-h.unregister:
			// 处理用户退出连接
			h.handleUnregister(client)
		case message := <-h.broadcast:
			// 处理发送广播消息
			for _, clients := range h.clients {
				for _, client := range clients {
					client.SendMessage(message)
				}
			}
		case <-h.closeSignal:
			// 处理 websocket 关闭
			Logger().Errorf(ctx, "websocket close")
			return
		}
	}
}

func (h *Hub) Ping() {
	defer func() {
		if r := recover(); r != nil {
			Logger().Errorf(ctx, "ping err：%+v", string(debug.Stack()))
			return
		}
	}()

	// 定时任务处理超时心跳连接的客户端
	_, _ = gcron.AddSingleton(ctx, "0 */5 * * * *", func(ctx context.Context) {
		h.clearTimeoutConnections()
	})
}

// 关闭超时心跳连接的客户端
func (h *Hub) clearTimeoutConnections() {
	//Logger().Info(ctx, "clearTimeoutConnections")
	currentTime := gtime.Now()
	for _, clients := range h.clients {
		for _, client := range clients {
			if currentTime.After(client.heartbeatTimeout) {
				//Logger().Errorf(ctx, "clearTimeoutConnections：[%v],currentTime=[%s],connTime=[%s],heartbeatTimeout=[%s]", client.id, currentTime, client.connTime, client.heartbeatTimeout)
				h.handleUnregister(client)
			}
		}
	}
}

func ForceClientLogout(clientID interface{}) {
	for _, clients := range hubManager.clients {
		for _, client := range clients {
			if client.id == clientID {
				Logger().Errorf(ctx, "ForceClientLogout：[%v]", client.id)
				client.close()
			}
		}
	}
}

func OnlineClientList() interface{} {
	clientList := make([]interface{}, 0)
	for _, clients := range hubManager.clients {
		for _, client := range clients {
			broswer, _ := client.userAgent.Browser()
			clientList = append(clientList, g.Map{
				"id": client.user.Id,
				"conn": g.Map{
					"remoteAddr": client.conn.RemoteAddr(),
					"localAddr":  client.conn.LocalAddr(),
				},
				"sendLen":          len(client.send),
				"sendClosed":       client.sendClosed,
				"connTime":         client.connTime.Time.Format(time.DateTime),
				"heartbeatTimeout": client.heartbeatTimeout.Time.Format(time.DateTime),
				"user":             client.user,
				"userAgent": g.Map{
					"os":      client.userAgent.OS(),
					"browser": broswer,
					"ua":      client.userAgent.UA(),
				},
			})
		}
	}
	return clientList
}

func HubBaseInfo() interface{} {
	return g.Map{
		"clients":       OnlineClientList(),
		"registerLen":   len(hubManager.register),
		"unregisterLen": len(hubManager.unregister),
		"broadcastLen":  len(hubManager.broadcast),
	}
}
